package com.lmind.core.message;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.mina.core.service.IoService;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import com.lmind.bean.EventProvider;
import com.lmind.core.async.ResultHandler;

public class MinaDuplexMessageService implements DuplexMessageService, MessageServiceListener, RequestResponseProcessor {
	
	private MessageCodec codec;
	
	private IoService ioService;
	
	private MessagePipeIoHandler ioHandler = new MessagePipeIoHandler();
	
	private MessagePipeIoHandlerConfig config = new MessagePipeIoHandlerConfig();
	
	private volatile PipelineManager messageHelper = new PipelineManager();
	
	private AsyncMessageProcessor<DuplexMessagePipe, Object> messageProcessor;
	
	private EventProvider<DuplexMessageServiceListener> event = new EventProvider<DuplexMessageServiceListener>();
	
	private ScheduledExecutorService scheduledExecutor;

	public void setConfig(MessagePipeIoHandlerConfig config) {
		this.config = config;
	}

	public void setCodec(MessageCodec codec) {
		this.codec = codec;
	}

	public void setMessageProcessor(
			AsyncMessageProcessor<DuplexMessagePipe, Object> messageProcessor) {
		this.messageProcessor = messageProcessor;
	}
	
	public void setScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
		this.scheduledExecutor = scheduledExecutor;
	}
	
	public void bindIoService(IoService ioService) {
		
		if (this.ioService != null)
			throw new MessagePipeException("服务已绑定");
		
		this.ioService = ioService;
		ioService.setHandler(ioHandler);
		
		
		messageHelper.setMessagePipe(ioHandler);
		messageHelper.setMessageProcessor(this);
		messageHelper.setExecutorService(scheduledExecutor);
		messageHelper.init();
		
		ioHandler.setIoService(ioService);
		ioHandler.setConfig(config);
		ioHandler.setCodec(codec);
		ioHandler.init();

		// 此代码必须在messageHelper的初始化之后
		ioHandler.addMessageServiceListener(this);
	}

	private void bind(InetSocketAddress address) throws IOException {
		
		NioSocketAcceptor acceptor = new NioSocketAcceptor();
		
		bindIoService(acceptor);
		
		acceptor.bind(address);
	}
	
	private void connect(InetSocketAddress address) {

		NioSocketConnector connector = new NioSocketConnector();
		
		bindIoService(connector);
		
		if (!connector.connect(address).awaitUninterruptibly(3000))
			throw new MessagePipeException("无法连接服务器");
	}
	
	public void close() {
		ioService.dispose(true);
	}
	
	
	@Override
	public void addDuplexMessageServiceListener(
			DuplexMessageServiceListener listener) {
		event.addEventListener(listener);
	}

	@Override
	public void removeDuplexMessageServiceListener(
			DuplexMessageServiceListener listener) {
		event.removeEventListener(listener);
	}

	@Override
	public List<DuplexMessagePipe> getPipes() {
		
		List<DuplexMessagePipe> list = new ArrayList<DuplexMessagePipe>();
		for (MessagePipe pipe : ioHandler.getMessagePipes()) {
			Pipe p = (Pipe)pipe.getAttribute(Pipe.class.getName());
			if (p != null)
				list.add(p);
		}
		return list;
	}

	@Override
	public void connected(MessagePipe pipe) {
		Pipe p = new Pipe(pipe);
		pipe.setAttribute(Pipe.class.getName(), p);
		
		for (DuplexMessageServiceListener listener : event.toList()) {
			listener.connected(p);
		}
	}

	@Override
	public void disconnected(MessagePipe pipe) {
		Pipe p = (Pipe)pipe.getAttribute(Pipe.class.getName());
		for (DuplexMessageServiceListener listener : event.toList()) {
			listener.disconnected(p);
		}
	}

	@Override
	public void process(MessagePipe context, Object message,
			ResultHandler<Object> handler) {
		
		Pipe p = (Pipe)context.getAttribute(Pipe.class.getName());
		
		if (message instanceof Request) {
			class Handler implements ResultHandler<Object> {
				private ResultHandler<Object> handler;
				public Handler(ResultHandler<Object> handler) {
					this.handler = handler;
				}
				public void result(Object result) {
					Response response = new Response(result);
					handler.result(response);
				}
				public void exception(Throwable e) {
					handler.exception(e);
				}
			}
			messageProcessor.process(p, ((Request)message).getMessage(), new Handler(handler));
		} else if (message instanceof KeepAlive) {
			message.toString();
		} else if (message instanceof Command) {
			messageProcessor.process(p, ((Command)message).getMessage(), null);
		} else {
			throw new MessagePipeException("未知的消息");
		}
	}

	@Override
	public boolean isRequest(MessagePipe pipe, Object message) {
		return message instanceof Request;
	}

	@Override
	public boolean isResponse(MessagePipe pipe, Object message) {
		return message instanceof Response;
	}

	@Override
	public void message(MessagePipe pipe, Object message) {
	}

	@Override
	public void keepAlive(MessagePipe pipe) {
		pipe.send(new KeepAlive());
	}
	
	
	class Pipe implements DuplexMessagePipe {
		
		private MessagePipe pipe;
		
		private AsyncMessageProcessor<DuplexMessagePipe, Object> processor;
		
		private Map<Object, Object> attributes = new ConcurrentHashMap<Object, Object>();

		public Pipe(MessagePipe pipe) {
			this.pipe = pipe;
		}

		@Override
		public void send(Object message) {
			if (message == null)
				throw new IllegalArgumentException("参数不可是null");
			pipe.send(new Command(message));
		}
		
		@Override
		public void close() {
			pipe.close();
		}

		@Override
		public Object getAttribute(Object key) {
			return attributes.get(key);
		}

		@Override
		public void setAttribute(Object key, Object value) {
			attributes.put(key, value);
		}

		@Override
		public void removeAttribute(Object key) {
			attributes.remove(key);
		}
		
		@Override
		public void setAsyncMessageProcessor(
				AsyncMessageProcessor<DuplexMessagePipe, Object> processor) {
			this.processor = processor;
		}

		@Override
		public void request(Object message, ResultHandler<Object> handler) {
			
			if (message == null || handler == null)
				throw new IllegalArgumentException("参数不可是null");
			
			class Handler implements ResultHandler<Object> {
				private ResultHandler<Object> handler;
				public Handler(ResultHandler<Object> handler) {
					this.handler = handler;
				}
				public void result(Object result) {
					Message response = (Message)result;
					handler.result(response.getMessage());
				}
				public void exception(Throwable e) {
					handler.exception(e);
				}
			}
			
			Request request = new Request(message);
			messageHelper.send(pipe, request, new Handler(handler));
		}
		
	}

}
